package org.yuanzheng.state

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector
import org.yuanzheng.source.StationLog

/**
 * 统计每个手机的呼叫时间间隔
 */
object KeyedState1 {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val path = getClass.getResource("/station.log").getPath
    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.readTextFile(path).map(line => {
      val split = line.split(",")
      StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
    })
    //
    val result = stream.keyBy(_.callOut).flatMap(new CallIntervalFunction)
    result.print()
    streamEnv.execute()
  }
}

class CallIntervalFunction extends RichFlatMapFunction[StationLog, (String, Long)] {
  //定义一个状态，用于保存上一次呼叫时间

  private var preCallTimeState: ValueState[Long] = _

  override def open(parameters: Configuration): Unit = {
    preCallTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("pre", classOf[Long]))
  }

  override def flatMap(in: StationLog, collector: Collector[(String, Long)]): Unit = {
    // 从状态中获取上一次呼叫的时间
    val preCallTime = preCallTimeState.value()
    if (preCallTime == 0 || preCallTime == null) {
      preCallTimeState.update(in.callTime)
    } else {
      val interval = in.callTime - preCallTime
      collector.collect((in.callOut, interval))
    }
  }
}